package org.tinygame.herostory.mq;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tinygame.herostory.rank.RankService;

import java.util.List;

/**
 * @author anan
 * @create 2022-05-08 15:39
 */
public final class MqConsumer {
    static private final Logger LOGGER = LoggerFactory.getLogger(MqConsumer.class);
    
    private MqConsumer(){}
    
    static public void init(){

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("herostory");

        consumer.setNamesrvAddr("127.0.0.1:9876");

        try{
            // 订阅主题
            //
            // XXX 注意: 想要订阅 herostory_victor 主题,
            // 必须先保证这个主题是存在的!
            // 否则后面的监听逻辑都不会执行...
            consumer.subscribe("herostory_victory", "*");

            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgExtList, ConsumeConcurrentlyContext ctx) {
                    for(MessageExt msgExt : msgExtList){
                        if(null == msgExt) continue;

                        VictorMsg victorMsg = JSONObject.parseObject(
                                msgExt.getBody(),
                                VictorMsg.class
                        );

                        LOGGER.info(
                                "从消息队列中收到胜利消息, winnerId = {}, loserId = {}",
                                victorMsg.winnerId,
                                victorMsg.loserId
                        );

                        RankService.getInstance().refreshRank(
                                victorMsg.winnerId,
                                victorMsg.loserId
                        );

                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();

            LOGGER.info("消息队列 ( 消费者 ) 连接成功!");
        }catch (Exception e){
            LOGGER.error(e.getMessage(), e);
        }
    }
}
